package com.shujia.flink.tf

import org.apache.flink.streaming.api.scala._

object Demo2Agg {
  def main(args: Array[String]): Unit = {


    //创建flink的运行环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment


    val ds: DataStream[String] = env.readTextFile("flink/data/students.txt")


    val student: DataStream[Student] = ds.map(line => {
      val split: Array[String] = line.split(",")
      Student(split(0), split(1), split(2).toInt, split(3), split(4))
    })


    /**
      * max  返回最大的值   其它列可能不对了
      * maxBy 返回最大值的键   其他的还是对的
      *
      * min
      * minBy
      *
      * sum
      *
      * reduce
      *
      *
      */

    //统计同一个班级年龄最大的学生
    student
      .keyBy(_.clazz)
      .min("age")
      .print()

    env.execute()


  }

  case class Student(id: String, name: String, age: Int, gender: String, clazz: String)

}
